package cn.dmrliu.edu.realtime.app.dws;

import cn.dmrliu.edu.realtime.bean.TrafficPageViewBean;
import cn.dmrliu.edu.realtime.util.ClickhouseUtil;
import cn.dmrliu.edu.realtime.util.DateUtil;
import cn.dmrliu.edu.realtime.util.KafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 基本环境的搭建
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点的设置

        // TODO 3. 从kafka的页面事务事实读取数据
        String topic = "edu_dwd_traffic_page";
        String group = "edu_dws_traffic_page_view_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka source");

//        kafkaDS.print(">>>");


        // TODO 4. 转为json格式
        //{"common":{"
        //  sc":"1",
        //  "ar":"27",
        //  "uid":"42",
        //  "os":"Android 11.0",
        //  "ch":"oppo",
        //  "is_new":"0",
        //  "md":"Xiaomi 9",
        //  "mid":"mid_147",
        //  "vc":"v2.1.132",
        //  "ba":"Xiaomi",
        //  "sid":"61e0932e-ae31-4cb1-ad63-a53c7a11ebf0"
        // },
        // "page":{
        //  "page_id":"home",
        //  "during_time":6722
        //  },
        //  "ts":1680002699991}
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDS.map(JSON::parseObject);

        // TODO 5. 按mid分组，转为实体类（可统计）
        KeyedStream<JSONObject, String> midKeyedDS = jsonObjectDS.keyBy(in -> in.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<TrafficPageViewBean> beanDS = midKeyedDS.process(new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
            private ValueState<String> lastVisitDateState;
            private MapState<String, Integer> sidCountState;


            // 状态变量初始化及设置过期时间
            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> lastVisitDateVSD = new ValueStateDescriptor<>(
                        "last_visit_date",
                        String.class
                );
                // 设置过期时间,1天
                lastVisitDateVSD.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                lastVisitDateState = getRuntimeContext().getState(lastVisitDateVSD);

                MapStateDescriptor<String, Integer> sidCountVSD = new MapStateDescriptor<>(
                        "sid_count",
                        String.class,
                        Integer.class
                );
                sidCountVSD.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                sidCountState = getRuntimeContext().getMapState(sidCountVSD);
            }

            @Override
            public void processElement(JSONObject jsonObject, Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                JSONObject commonJsonObject = jsonObject.getJSONObject("common");
                JSONObject pageJsonObject = jsonObject.getJSONObject("page");
                Long ts = jsonObject.getLong("ts");

                // 获取维度数据
                String vc = commonJsonObject.getString("vc");
                String ch = commonJsonObject.getString("ch");
                String ar = commonJsonObject.getString("ar");
                String isNew = commonJsonObject.getString("is_new");

                long uvCt = 0L;
                long svCt = 0L;
                long ujCt = 0L;
                long pvCt = 1L;// 页面浏览数
                long durSum = pageJsonObject.getLong("during_time");// 访问时长

                // 独立访客数
                String lastVisitDate = lastVisitDateState.value();
                String curVisitDate = DateUtil.toDate(ts);
                if (StringUtils.isEmpty(lastVisitDate) || !lastVisitDate.equals(curVisitDate)) {
                    uvCt = 1L;
                    lastVisitDateState.update(curVisitDate);
                }
                // 会话数
                String lastPageId = pageJsonObject.getString("last_page_id");
                if (StringUtils.isEmpty(lastPageId)) {
                    svCt = 1L;
                }

                // 跳出会话数
                String sid = commonJsonObject.getString("sid");

                // 如果第一次出现，跳出1个
                // 如果第二次出现，跳出-1个
                // 如果第三次出现，跳出0个

                 // TODO 更改逻辑：如果上次是首页，本次也是首页，就是跳出一个
                if (!sidCountState.contains(sid)) {
                    sidCountState.put(sid, 1);
                    ujCt = 1L;
                } else {
                    if (sidCountState.get(sid) == 1) {
                        sidCountState.put(sid, 2);
                        ujCt = -1L;
                    }
                }

                out.collect(new TrafficPageViewBean(
                        "",
                        "",
                        vc,
                        ch,
                        ar,
                        isNew,
                        uvCt,
                        svCt,
                        ujCt,
                        pvCt,
                        durSum,
                        ts
                ));

            }
        });

        // TODO 6. 设置水位线
        SingleOutputStreamOperator<TrafficPageViewBean> watermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );

        // TODO 7. 按维度分组，开窗，统计
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = watermarkDS
                .keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4 getKey(TrafficPageViewBean in) throws Exception {
                        return Tuple4.of(in.getVc(), in.getCh(), in.getAr(), in.getIsNew());
                    }
                })
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(30)))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean v1, TrafficPageViewBean v2) throws Exception {
                                v1.setUvCt(v1.getUvCt() + v2.getUvCt());
                                v1.setSvCt(v1.getSvCt() + v2.getSvCt());
                                v1.setUjCt(v1.getUjCt() + v2.getUjCt());
                                v1.setPvCt(v1.getPvCt() + v2.getPvCt());
                                v1.setDurSum(v1.getDurSum() + v2.getDurSum());
//                                System.out.println("++++++++++++++++++++" + v1.getUjCt());
                                return v1;
                            }
                        },
                        new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void apply(Tuple4<String, String, String, String> tuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                                String stt = DateUtil.toYmdHms(window.getStart());
                                String edt = DateUtil.toYmdHms(window.getEnd());
                                for (TrafficPageViewBean bean : input) {
                                    bean.setStt(stt);
                                    bean.setEdt(edt);
                                    bean.setTs(System.currentTimeMillis());
                                    if (bean.getUjCt() < 0) {
                                        System.out.println(bean.getUjCt());
                                        bean.setUjCt(0L);
                                    }
                                    out.collect(bean);
                                }
                            }
                        });

        reduceDS.print(">>>>>>");
        // TODO 8.写入clickhouse
        reduceDS.addSink(ClickhouseUtil
                .getSinkFunction("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
